#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>

#include <sys/socket.h>
#include <sys/epoll.h>

#include <arpa/inet.h>
#include <netinet/in.h>
#include <pthread.h>

#include "network.h"
#include "sessionfactory.h"
#include "msgqueue.h"

static const int State_Unconnected = 0;
static const int State_Connected = 1;
static const int State_Closing = 2;
static const int State_Closed = 3;

static pthread_cond_t sleep_cond;
static pthread_mutex_t sleep_mutex;
static int sleepSwitch;

static AppSession *session_array;
static struct bufferFactory *buffer_factory;

//**********************************************************************
inline int is_closed(AppSession *session) {
	int closed = session->closed == State_Closed;
	int closing = session->closed == State_Closing;
	return closed | closing;
}

inline int send_msg(AppSession *session, struct buffer *msg) {
	if (is_closed(session)) {
		return 1;
	}
	ready_to_write(msg);
	appendToTail(session->responses, msg);

	return 0;

}

inline int close_session(AppSession *session) {
	session->closed = State_Closing;
	return 0;
}
//**************above of all are appsession.h****************************************************

struct _threadarg {
	Network *work;
	NetIO *io;
};
static inline int add_fd_to_epoll(struct epoll_event *ev, int epollfd, int fd,
		uint32_t event);
static inline int del_fd_from_epoll(int epollfd, int fd);
static void initial_session_array(Network *work);
static inline void reset_session(AppSession *session);

void *msg_thread(void *arg) {
	struct _threadarg *ta = (struct _threadarg*) arg;
	Network *network = ta->work;
	NetIO *io = ta->io;

	u_int32_t overload = network->overload;
	int sleepFlag;
	for (;;) {
		sleepFlag = 1;

		u_int32_t i = 3;
		for (; i < overload; i++) { //msg received
			AppSession *session = &(session_array[i]);
			if (session->closed == State_Closing) {
				session->closed = State_Closed;
				continue;
			}
			if (session->closed != State_Connected) {
				continue;
			}

			struct buffer *request = popFromHead(session->requests);
			if (request == NULL) {
				sleepFlag = 1;
			} else {
				network->received(session, request);
				sleepFlag = 0;
			}

		}

		i = 3;
		for (; i < overload; i++) { //msg sent
			AppSession *session = &(session_array[i]);
			if (session->closed == State_Closing) {
				session->closed = State_Closed;
				continue;
			}
			if (session->closed != State_Connected) {
				continue;
			}

			struct buffer *response = popFromHead(session->responses);
			if (response == NULL) {
				sleepFlag = 1;
			} else {
				network->sent(session, response);
				sleepFlag = 0;
			}
		}

		i = 3;
		for (; i < overload; i++) { //close connection
			if (State_Closing == session_array[i].closed) {
				sleepFlag = 0;
				session_array[i].closed = State_Closed;
				continue;
			}

			if (State_Closed != session_array[i].closed) {
				continue;
			}

			sleepFlag = 0;

			AppSession *session = &(session_array[i]);
			network->closed(session);

			close(session_array[i].fd);
			session_array[i].closed = State_Unconnected;
		}

		if (sleepFlag) {
			pthread_mutex_lock(&sleep_mutex);
			while (sleepSwitch == 0) {
				pthread_cond_wait(&sleep_cond, &sleep_mutex);
			}
			sleepSwitch = 0;
			pthread_mutex_unlock(&sleep_mutex);
		}
	}
	return "e";
}

void *io_thread(void *arg) {
	struct _threadarg *ta = (struct _threadarg*) arg;
	Network *network = ta->work;
	NetIO *io = ta->io;

	int listen_sock;
	int ret;
	listen_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (listen_sock < 0) {
		perror("socket failed");
		return "e";
	}
	network->listenfd = listen_sock;

	struct sockaddr_in client;
	socklen_t client_len = sizeof(client);

	ret = bind(listen_sock, network->addr, sizeof(struct sockaddr_in));
	if (ret < 0) {
		perror("bind failed");
		return "e";
	}

	ret = listen(listen_sock, network->overload);
	if (ret < 0) {
		perror("listen failed");
		return "e";
	}

	struct epoll_event ev;
	struct epoll_event events[network->overload];

	int conn_sock, nfds, epollfd;

	epollfd = epoll_create(network->overload);
	if (epollfd == -1) {
		perror("epoll_create");
		return "e";
	}

	ret = add_fd_to_epoll(&ev, epollfd, listen_sock, EPOLLIN);
	if (ret < 0) {
		perror("epoll_ctl");
		return "e";
	}

	initial_session_array(network);
	//TODO msg thread loop within session_Array that are not good,
	pthread_t msg_run;
	pthread_create(&msg_run, NULL, msg_thread, arg);

	while (network->running) {
		nfds = epoll_wait(epollfd, events, network->overload, -1);
		if (nfds == -1) {
			close(listen_sock);
			perror("epoll_pwait");
			return "e";
		}

		int n;
		for (n = 0; n < nfds; ++n) {
			if (events[n].data.fd == listen_sock) {
				conn_sock = accept(listen_sock, (struct sockaddr *) &client,
						&client_len);
				if (conn_sock == -1) {
					perror("accept");
					continue;
				}
				//TODO check network overload
				//setnonblocking(conn_sock);
				AppSession *session = &(session_array[conn_sock]);
				if (session->closed != State_Unconnected) {
					close(conn_sock);
					continue;
				}

				ret = add_fd_to_epoll(&ev, epollfd, conn_sock, EPOLLIN);
				if (ret < 0) {
					session->closed = State_Unconnected;
					perror("epoll_ctl: conn_sock");
					close(conn_sock);
					continue;
				}
				session->fd = conn_sock;
				reset_session(session); //TODO new iosession
				network->connected(session, (struct sockaddr *) &client);
				continue;
			} //end listen socket

			int fd = events[n].data.fd; //TODO find iosession from active sessions
			AppSession *session = &(session_array[fd]);
			if (session->closed == State_Closing) {
				session->closed = State_Closed;
				continue;
			}
			if (session->closed != State_Connected) {
				continue;
			}
			struct buffer *pbuff = session->buff;
			u_int32_t index = pbuff->index;
			u_int32_t limit = pbuff->limit;
			u_int32_t free_space = limit - index;
			int ret_read = read(session->fd, (pbuff->data) + index, free_space);
			if (ret_read == -1) {
				if (errno == EAGAIN) {
					continue;
				}
				if (errno == EINTR) {
					continue;
				}
				del_fd_from_epoll(epollfd, fd);
				session->closed = State_Closed;
				continue;
			}

			if (ret_read == 0) {
				del_fd_from_epoll(epollfd, fd);
				session->closed = State_Closed;
				continue;
			}
			pbuff->index = (pbuff->index) + ret_read;
			ready_to_write(session->buff);

			io->decode(session, session->buff); //TODO batch add frame msg to queue
			buffer_compact(session->buff);

		} // end of read data from socket

		for (n = 0; n < nfds; ++n) {
			int fd = events[n].data.fd;
			AppSession *session = &(session_array[fd]);
			if (session->closed == State_Closing) {
				session->closed = State_Closed;
				continue;
			}
			if (session->closed != State_Connected) {
				continue;
			}

			struct buffer *response = topOfQueue(session->responses);
			if (response == NULL) {
				continue;
			}
			u_int32_t index = response->index;
			u_int32_t limit = response->limit;
			u_int32_t data_size = limit - index;
			int ret_write = write(session->fd, (response->data) + index,
					data_size);
			if (ret_write == -1) {
				if (errno != EAGAIN && errno != EINTR) {
					del_fd_from_epoll(epollfd, fd);
					session->closed = State_Closed;
					continue;
				}
			}

			if (ret_write == data_size) {
				response = popFromHead(session->responses);
				network->sent(session, response);
			}

		} // end of write data

		pthread_mutex_lock(&sleep_mutex);
		sleepSwitch = 1;
		pthread_cond_signal(&sleep_cond);
		pthread_mutex_unlock(&sleep_mutex);
		

	} // end of io loop
	pthread_join(msg_run, NULL);

	del_fd_from_epoll(epollfd, listen_sock);
	close(listen_sock);
	return NULL;
}

int start_work(Network *network, NetIO *io) {

	struct _threadarg arg;
	arg.io = io;
	arg.work = network;

	pthread_mutex_init(&sleep_mutex, NULL);
	pthread_cond_init(&sleep_cond, NULL);
	sleepSwitch = 1;

	pthread_t io_run;
	pthread_create(&io_run, NULL, io_thread, (void*) &arg);
	pthread_join(io_run, NULL);

	return 0;
}

int stop_work(Network *network) {
	network->running = 0;
	return 0;
}

inline int add_fd_to_epoll(struct epoll_event *ev, int epollfd, int fd,
		uint32_t event) {
	(*ev).events = event;
	(*ev).data.fd = fd;
	return epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, ev);
}

inline int del_fd_from_epoll(int epollfd, int fd) {
	return epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, NULL);
}

static void initial_session_array(Network *work) {
	size_t session_len = sizeof(struct appSession);
	session_array = malloc(session_len * (work->overload));

	buffer_factory = bufferFactory_build(4096, work->overload);
	u_int32_t i = 0;
	for (; i < work->overload; i++) {
		session_array[i].fd = -1;
		session_array[i].closed = State_Unconnected;
		session_array[i].buff = buffer_new(buffer_factory);
		session_array[i].requests = newQueue(1);
		session_array[i].responses = newQueue(1);
		ready_to_read(session_array[i].buff);
	}
}

inline void reset_session(AppSession *session) {
	session->closed = State_Connected;

}

